package com.boarsoft.message.sofamq;

import java.io.Closeable;
import java.io.IOException;
import java.util.Properties;

import io.openmessaging.api.Action;
import io.openmessaging.api.ConsumeContext;
import io.openmessaging.api.Consumer;
import io.openmessaging.api.Message;
import io.openmessaging.api.MessageListener;
import io.openmessaging.api.OMS;

public class SofaMQConsumer implements MessageListener, Closeable {
	protected String groupId = "default";

	protected SofaMqConfig sofaMqConfig;

	protected Consumer consumer;

	protected MessageListener listener;

	protected String tag = "*";

	public void init(String code) {
		Properties props = new Properties();
		props.setProperty("groupId", this.groupId);
		props.setProperty("instanceId", this.sofaMqConfig.getInstanceId());
		consumer = OMS.builder().driver("sofamq")//
				.build(this.sofaMqConfig.getBaseMqProperties())//
				.createConsumer(props);
		consumer.start();
		consumer.subscribe(code, tag, this);
	}

	@Override
	public Action consume(Message m, ConsumeContext cc) {
		if (listener != null) {
			listener.consume(m, cc);
			return Action.CommitMessage;
		}
		return Action.ReconsumeLater;
	}

	@Override
	public void close() throws IOException {

	}

	public String getGroupId() {
		return groupId;
	}

	public void setGroupId(String groupId) {
		this.groupId = groupId;
	}

	public SofaMqConfig getSofaMqConfig() {
		return sofaMqConfig;
	}

	public void setSofaMqConfig(SofaMqConfig sofaMqConfig) {
		this.sofaMqConfig = sofaMqConfig;
	}

	public Consumer getConsumer() {
		return consumer;
	}

	public void setConsumer(Consumer consumer) {
		this.consumer = consumer;
	}

	public MessageListener getListener() {
		return listener;
	}

	public void setListener(MessageListener listener) {
		this.listener = listener;
	}

	public String getTag() {
		return tag;
	}

	public void setTag(String tag) {
		this.tag = tag;
	}

}
